package com.tool.rabbitmq;

import com.tool.rabbitmq.constants.RabbitConstant;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;


/**
 * @author : zhaoxuan
 * @date : 2021/9/10 13:32
 * @description : RabbitAutoService
 * @version: 1.0
 */
public class RabbitAutoService{
    private final String EMPTY = "";
    private static String X_DELAY_HEADER = "x-delay";

    private RabbitTemplate rabbitTemplate;
    private RabbitAdmin rabbitAdmin;
    private CustomExchange delayExchange;
    private ApplicationContext applicationContext;

    public RabbitAutoService(RabbitTemplate rabbitTemplate, RabbitAdmin rabbitAdmin, CustomExchange delayExchange, ApplicationContext applicationContext) {
        this.rabbitTemplate = rabbitTemplate;
        this.rabbitAdmin = rabbitAdmin;
        this.delayExchange = delayExchange;
        this.applicationContext = applicationContext;
    }

    /**
     * 发送SIMPLE MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(queuesToDeclare = @Queue("queue"))
     */
    public void sendSimpleMessage(String queue, Object message) {
        rabbitAdmin.declareQueue(QueueBuilder.durable(queue).build());
        sendSingleMessage(EMPTY, queue, message, null);
    }

    public void sendSimpleMessage(String queue, Object message, CorrelationData correlationData) {
        rabbitAdmin.declareQueue(QueueBuilder.durable(queue).build());
        sendSingleMessage(EMPTY, queue, message, correlationData);
    }

    public void sendSimpleMessage(String queue, Object message,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        rabbitAdmin.declareQueue(QueueBuilder.durable(queue).build());
        sendPrototypeMessage(EMPTY, queue, message, null, confirmCallback, returnCallback);
    }

    public void sendSimpleMessage(String queue, Object message, CorrelationData correlationData,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        rabbitAdmin.declareQueue(QueueBuilder.durable(queue).build());
        sendPrototypeMessage(EMPTY, queue, message, correlationData, confirmCallback, returnCallback);
    }
    /**发送SIMPLE MESSAGE------------------------------END--------------------------------------------*/

    /**
     * 发送DEFAULT DIRECT MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = RabbitConstant.DEFAULT_DIRECT, type = ExchangeTypes.DIRECT),
     * key = "routingKey")})
     */
    public void sendDefaultDirectMessage(String queue, Object message) {
        createDirectQueueExchangeBuilding(RabbitConstant.DEFAULT_DIRECT, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_DIRECT, queue, message, null);
    }

    public void sendDefaultDirectMessage(String queue, Object message, CorrelationData correlationData) {
        createDirectQueueExchangeBuilding(RabbitConstant.DEFAULT_DIRECT, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_DIRECT, queue, message, correlationData);
    }

    public void sendDefaultDirectMessage(String queue, Object message,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createDirectQueueExchangeBuilding(RabbitConstant.DEFAULT_DIRECT, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_DIRECT, queue, message, null, confirmCallback, returnCallback);
    }

    public void sendDefaultDirectMessage(String queue, Object message, CorrelationData correlationData,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createDirectQueueExchangeBuilding(RabbitConstant.DEFAULT_DIRECT, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_DIRECT, queue, message, correlationData, confirmCallback, returnCallback);
    }
    /**发送DEFAULT DIRECT MESSAGE------------------------------END--------------------------------------------*/

    /**
     * 发送DIRECT MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = exchange, type = ExchangeTypes.DIRECT),
     * key = "routingKey")})
     */
    public void sendDirectMessage(String exchange, String queue, Object message) {
        createDirectQueueExchangeBuilding(exchange, queue);
        sendSingleMessage(exchange, queue, message, null);
    }

    public void sendDirectMessage(String exchange, String queue, Object message, CorrelationData correlationData) {
        createDirectQueueExchangeBuilding(exchange, queue);
        sendSingleMessage(exchange, queue, message, correlationData);
    }

    public void sendDirectMessage(String exchange, String queue, Object message,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createDirectQueueExchangeBuilding(exchange, queue);
        sendPrototypeMessage(exchange, queue, message, null, confirmCallback, returnCallback);
    }

    public void sendDirectMessage(String exchange, String queue, Object message, CorrelationData correlationData,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createDirectQueueExchangeBuilding(exchange, queue);
        sendPrototypeMessage(exchange, queue, message, correlationData, confirmCallback, returnCallback);
    }

    private void createDirectQueueExchangeBuilding(String exchange, String queue) {
        Queue queue_ = QueueBuilder.durable(queue).build();
        rabbitAdmin.declareQueue(queue_);
        Exchange exchange_ = ExchangeBuilder.directExchange(exchange).durable(true).build();
        rabbitAdmin.declareExchange(exchange_);
        Binding binding = BindingBuilder.bind(queue_).to(exchange_).with(queue).noargs();
        rabbitAdmin.declareBinding(binding);
    }
    /**发送DIRECT MESSAGE------------------------------END--------------------------------------------*/


    /**
     * 发送DEFAULT FANOUT MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = RabbitConstant.DEFAULT_FANOUT, type = ExchangeTypes.FANOUT))})
     */
    public void sendDefaultFanoutMessage(String queue, Object message) {
        createFanoutQueueExchangeBuilding(RabbitConstant.DEFAULT_FANOUT, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_FANOUT, EMPTY, message, null);
    }

    public void sendDefaultFanoutMessage(String queue, Object message, CorrelationData correlationData) {
        createFanoutQueueExchangeBuilding(RabbitConstant.DEFAULT_FANOUT, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_FANOUT, EMPTY, message, correlationData);
    }

    public void sendDefaultFanoutMessage(String queue, Object message, RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createFanoutQueueExchangeBuilding(RabbitConstant.DEFAULT_FANOUT, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_FANOUT, EMPTY, message, null, confirmCallback, returnCallback);
    }

    public void sendDefaultFanoutMessage(String queue, Object message, CorrelationData correlationData,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createFanoutQueueExchangeBuilding(RabbitConstant.DEFAULT_FANOUT, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_FANOUT, EMPTY, message, correlationData, confirmCallback, returnCallback);
    }
    /**发送DEFAULT FANOUT MESSAGE------------------------------END--------------------------------------------*/

    /**
     * 发送FANOUT MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = "exchange", type = ExchangeTypes.FANOUT))})
     */
    public void sendFanoutMessage(String exchange, String queue, Object message) {
        createFanoutQueueExchangeBuilding(exchange, queue);
        sendSingleMessage(exchange, EMPTY, message, null);
    }

    public void sendFanoutMessage(String exchange, String queue, Object message, CorrelationData correlationData) {
        createFanoutQueueExchangeBuilding(exchange, queue);
        sendSingleMessage(exchange, EMPTY, message, correlationData);
    }

    public void sendFanoutMessage(String exchange, String queue, Object message,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createFanoutQueueExchangeBuilding(exchange, queue);
        sendPrototypeMessage(exchange, EMPTY, message, null, confirmCallback, returnCallback);
    }

    public void sendFanoutMessage(String exchange, String queue, Object message,
                                  CorrelationData correlationData,
                                  RabbitTemplate.ConfirmCallback confirmCallback,
                                  RabbitTemplate.ReturnCallback returnCallback) {
        createFanoutQueueExchangeBuilding(exchange, queue);
        sendPrototypeMessage(exchange, EMPTY, message, correlationData, confirmCallback, returnCallback);
    }

    private void createFanoutQueueExchangeBuilding(String exchange, String queue) {
        Queue queue_ = QueueBuilder.durable(queue).build();
        rabbitAdmin.declareQueue(queue_);
        Exchange exchange_ = ExchangeBuilder.fanoutExchange(exchange).durable(true).build();
        rabbitAdmin.declareExchange(exchange_);
        Binding binding = BindingBuilder.bind(queue_).to(exchange_).with(EMPTY).noargs();
        rabbitAdmin.declareBinding(binding);
    }
    /**发送FANOUT MESSAGE------------------------------END--------------------------------------------*/

    /**
     * 发送Default TOPIC MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = RabbitConstant.DEFAULT_TOPIC, type = ExchangeTypes.TOPIC),
     * key = "routingKey")})
     */
    public void sendDefaultTopicMessage(String routingKey, String queue, Object message) {
        createTopicQueueExchangeBuilding(RabbitConstant.DEFAULT_TOPIC, routingKey, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_TOPIC, routingKey, message, null);
    }

    public void sendDefaultTopicMessage(String routingKey, String queue, Object message, CorrelationData correlationData) {
        createTopicQueueExchangeBuilding(RabbitConstant.DEFAULT_TOPIC, routingKey, queue);
        sendSingleMessage(RabbitConstant.DEFAULT_TOPIC, routingKey, message, correlationData);
    }

    public void sendDefaultTopicMessage(String routingKey, String queue, Object message,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createTopicQueueExchangeBuilding(RabbitConstant.DEFAULT_TOPIC, routingKey, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_TOPIC, routingKey, message, null, confirmCallback, returnCallback);
    }

    public void sendDefaultTopicMessage(String routingKey, String queue, Object message, CorrelationData correlationData,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createTopicQueueExchangeBuilding(RabbitConstant.DEFAULT_TOPIC, routingKey, queue);
        sendPrototypeMessage(RabbitConstant.DEFAULT_TOPIC, routingKey, message, correlationData, confirmCallback, returnCallback);
    }
    /**发送Default TOPIC MESSAGE------------------------------END--------------------------------------------*/

    /**
     * 发送TOPIC MESSAGE------------------------------START--------------------------------------------
     * 消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = exchange, type = ExchangeTypes.TOPIC),
     * key = "routingKey")})
     */
    public void sendTopicMessage(String exchange, String routingKey, String queue, Object message) {
        createTopicQueueExchangeBuilding(exchange, routingKey, queue);
        sendSingleMessage(exchange, routingKey, message, null);
    }

    public void sendTopicMessage(String exchange, String routingKey, String queue, Object message, CorrelationData correlationData) {
        createTopicQueueExchangeBuilding(exchange, routingKey, queue);
        sendSingleMessage(exchange, routingKey, message, correlationData);
    }

    public void sendTopicMessage(String exchange, String routingKey, String queue, Object message,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createTopicQueueExchangeBuilding(exchange, routingKey, queue);
        sendPrototypeMessage(exchange, routingKey, message, null, confirmCallback, returnCallback);
    }

    public void sendTopicMessage(String exchange, String routingKey, String queue, Object message, CorrelationData correlationData,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createTopicQueueExchangeBuilding(exchange, routingKey, queue);
        sendPrototypeMessage(exchange, routingKey, message, correlationData, confirmCallback, returnCallback);
    }

    private void createTopicQueueExchangeBuilding(String exchange, String routingKey, String queue) {
        Queue queue_ = QueueBuilder.durable(queue).build();
        rabbitAdmin.declareQueue(queue_);
        Exchange exchange_ = ExchangeBuilder.fanoutExchange(exchange).durable(true).build();
        rabbitAdmin.declareExchange(exchange_);
        Binding binding = BindingBuilder.bind(queue_).to(exchange_).with(routingKey).noargs();
        rabbitAdmin.declareBinding(binding);
    }
    /**发送TOPIC MESSAGE------------------------------END--------------------------------------------*/


    /**
     * 发送DELAY MESSAGE------------------------------START--------------------------------------------
     *  消费端使用示例
     * @RabbitListener(bindings = {
     * @QueueBinding(value = @Queue(value = "queue", durable = "true"),
     * exchange = @Exchange(value = RabbitConstant.DELAY_EXCHANGE),
     * key = "queue")})
     */
    public void sendDelayMessage(String queue, Object message, long delayTimeMs) {
        createDelayQueueExchangeBuilding(queue);
        sendSingleDelayMessage(queue, message, delayTimeMs, null);
    }

    public void sendDelayMessage(String queue, Object message, long delayTimeMs, CorrelationData correlationData) {
        createDelayQueueExchangeBuilding(queue);
        sendSingleDelayMessage(queue, message, delayTimeMs, correlationData);
    }

    public void sendDelayMessage(String queue, Object message, long delayTimeMs,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createDelayQueueExchangeBuilding(queue);
        sendPrototypeDelayMessage(queue, message, delayTimeMs, null, confirmCallback, returnCallback);
    }

    public void sendDelayMessage(String queue, Object message, long delayTimeMs, CorrelationData correlationData,
                                 RabbitTemplate.ConfirmCallback confirmCallback,
                                 RabbitTemplate.ReturnCallback returnCallback) {
        createDelayQueueExchangeBuilding(queue);
        sendPrototypeDelayMessage(queue, message, delayTimeMs, correlationData, confirmCallback, returnCallback);
    }

    private void createDelayQueueExchangeBuilding(String queue) {
        Queue queue_ = QueueBuilder.durable(queue).build();
        rabbitAdmin.declareQueue(queue_);
        Binding binding = BindingBuilder.bind(queue_).to(delayExchange).with(queue).noargs();
        rabbitAdmin.declareBinding(binding);
    }

    private void sendSingleMessage(String exchange,
                                   String routingKey,
                                   Object message,
                                   CorrelationData correlationData) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
    }


    private void sendPrototypeMessage(String exchange,
                                      String routingKey,
                                      Object message,
                                      CorrelationData correlationData,
                                      RabbitTemplate.ConfirmCallback confirmCallback,
                                      RabbitTemplate.ReturnCallback returnCallback) {
        RabbitTemplate rabbitTemplate = getRabbitTemplate(confirmCallback, returnCallback);
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
    }

    private void sendSingleDelayMessage(String routingKey, Object message, long delayTimeMs, CorrelationData correlationData) {
        rabbitTemplate.convertAndSend(
                RabbitConstant.DELAY_EXCHANGE, routingKey, message,
                msg -> {
                    if (delayTimeMs >= 0) {
                        msg.getMessageProperties().setHeader(X_DELAY_HEADER, delayTimeMs);
                    } else {
                        msg.getMessageProperties().getHeaders().remove(X_DELAY_HEADER);
                    }
                    return msg;
                }, correlationData);
    }

    private void sendPrototypeDelayMessage(String routingKey, Object message, long delayTimeMs, CorrelationData correlationData,
                                           RabbitTemplate.ConfirmCallback confirmCallback,
                                           RabbitTemplate.ReturnCallback returnCallback) {
        RabbitTemplate rabbitTemplate = getRabbitTemplate(confirmCallback, returnCallback);
        rabbitTemplate.convertAndSend(
                RabbitConstant.DELAY_EXCHANGE, routingKey, message,
                msg -> {
                    if (delayTimeMs >= 0) {
                        msg.getMessageProperties().setHeader(X_DELAY_HEADER, delayTimeMs);
                    } else {
                        msg.getMessageProperties().getHeaders().remove(X_DELAY_HEADER);
                    }
                    return msg;
                }, correlationData);
    }

    private RabbitTemplate getRabbitTemplate(RabbitTemplate.ConfirmCallback confirmCallback, RabbitTemplate.ReturnCallback returnCallback) {
        RabbitTemplate rabbitTemplate;
        if (confirmCallback == null && returnCallback == null) {
            rabbitTemplate = this.rabbitTemplate;
        } else {
            rabbitTemplate = getRabbitTemplate();
            if (confirmCallback != null) {
                rabbitTemplate.setConfirmCallback(confirmCallback);
            }
            if (returnCallback != null) {
                rabbitTemplate.setReturnCallback(returnCallback);
            }
        }
        return rabbitTemplate;
    }

    private RabbitTemplate getRabbitTemplate() {
        return (RabbitTemplate) applicationContext.getBean(RabbitConstant.CALLBACK_RABBIT_TEMPLATE);
    }
}
